package urlcounter;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.ArrayWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import urlcounter.Stage1MapReduce.DuplicationRemoverReducer;
import urlcounter.Stage1MapReduce.TupleExtractorMapper;
import urlcounter.Stage2MapReduce.TupleCounterReducer;
import urlcounter.Stage2MapReduce.TupleCreatorMapper;


public class URLCount {
	
	public static class MyArrayWritable extends ArrayWritable {

		public MyArrayWritable(String[] strings) {
			super(strings);
		}

		public MyArrayWritable() {
			super(Text.class);
		}

		public String toString() {
			StringBuilder sb = new StringBuilder();
			for (String s : toStrings()) {
				sb.append(s + " ");
			}
			return sb.toString();
		}
	}

	public static void main(String[] args) throws Exception {
	
		Configuration conf1 = new Configuration();
		Configuration conf2 = new Configuration();
		Job job1 = new Job(conf1, "Collect set of urls for query id");
		Job job2 = new Job(conf2, "Count url touples appearences");
		
		Path inputPath = new Path("input/largeInput.txt");
		Path outputPathPhase1 = new Path("outputP1");
		Path outputPathPhase2 = new Path("outputP2");
		phase1(job1, inputPath, outputPathPhase1);
		phase2(job2, outputPathPhase1, outputPathPhase2);
	}

	private static int phase1(Job job, Path inputPath, Path outputPath) throws IOException, InterruptedException,
			ClassNotFoundException {
		//settings
		job.setJarByClass(URLCount.class);
		job.setMapperClass(TupleExtractorMapper.class);
		job.setReducerClass(DuplicationRemoverReducer.class);
		// Map output
		job.setMapOutputKeyClass(LongWritable.class);
		job.setMapOutputValueClass(Text.class);
		// Reduce output		
		job.setOutputKeyClass(LongWritable.class);
		job.setOutputValueClass(MyArrayWritable.class);
		//input 
		job.setInputFormatClass(TextInputFormat.class);
		//format
		FileInputFormat.addInputPath(job, inputPath);	
		FileOutputFormat.setOutputPath(job, outputPath);
		
		return (job.waitForCompletion(true) ? 0 : 1);
	}

	private static int phase2(Job job, Path inputPath, Path outputPath) throws IOException, InterruptedException,
	ClassNotFoundException {
		//settings
		job.setJarByClass(URLCount.class);
		job.setMapperClass(TupleCreatorMapper.class);
		job.setReducerClass(TupleCounterReducer.class);
		// Map output
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(LongWritable.class);
		// Reduce output		
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(LongWritable.class);
		//input 
		job.setInputFormatClass(TextInputFormat.class);
		//format
		FileInputFormat.addInputPath(job, inputPath);
		FileOutputFormat.setOutputPath(job, outputPath);
		
		return (job.waitForCompletion(true) ? 0 : 1);

	}

}
